package avicit.bdp.dcs.json.service;

import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.constant.ProcessConstants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dds.api.ProcessDefinitionApi;
import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.platform6.core.exception.BusinessException;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.http.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author LIYB json构建实现类
 */
@Service
@Transactional
public class TaskJsonService {

  private static final Logger logger = LoggerFactory.getLogger(TaskJsonService.class);

  public static final int MAX_RANDOM = 10000000;
  public static final String CREATE_TABLE = "createTable";
  public static final String GET_FILE_NODE = "getFileNode";
  public static final String PRE_PROCESS_NODE = "preProcessNode";
  public static final String CREATE_TABLE_NODE = "createTableNode";
  private static final String TYPE = "type";
  // 默认同时执行3个采集任务
  public static final double COLLECT_NUM = 3d;

  @Value("${dds.createTableUrl}")
  public String createTableUrl;

  @Value("${dds.getSyncFileUrl}")
  public String getSyncFileUrl;

  @Value("${dds.preProcessUrl}")
  public String preProcessUrl;

  @Resource private DcsJobService jobService;
  @Resource private ProcessDefinitionApi processDefinitionApi;
  @Resource private StructureTaskJsonService structureTaskJsonService;
  @Resource private FileTaskJsonService fileTaskJsonService;
  @Resource private StreamTaskJsonService streamTaskJsonService;
  @Resource private ApiTaskJsonService apiTaskJsonService;

  /** 保存DS流程定义 */
  public Integer saveProcessDefinition(String jobId) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    // 1、构建Dolphinscheduler流程定义json数据和流程定义节点连接信息json
    Map<String, String> jsonMap = new HashMap<>();
    if (jobDTO.getJobType() == JobType.TRANSFER.getCode()
        || jobDTO.getJobType() == JobType.SYNC.getCode()) {
      // 应用数据采集
      jsonMap = structureTaskJsonService.buildDsJson(jobDTO);
    } else if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      // 文件采集
      jsonMap = fileTaskJsonService.buildDsJson(jobDTO);
    } else if (jobDTO.getJobType() == JobType.STREAM.getCode()) {
      // 流式采集
      jsonMap = streamTaskJsonService.buildDsJson(jobDTO);
    } else if (jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
      // API采集
      jsonMap = apiTaskJsonService.buildDsJson(jobDTO);
    } else {
      logger.warn(DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + jobDTO.getJobType());
    }

    // 2、调用DS服务保存流程定义
    Map<String, Object> definitionMap = new HashMap<>();
    definitionMap.put(ProcessConstants.ID, jobId);
    definitionMap.put(ProcessConstants.PROJECT_ID, jobDTO.getProjectId());
    definitionMap.put(ProcessConstants.TYPE_ID, TaskType.DATAX.getDescription());

    // 类型：0--应用数据采集；1--数据迁移；2--数据流程开发；3--文件采集;4--流式采集；5--API采集； 6--开发工作台；7--离线计算；8--实时计算；9--机器学习
    if (jobDTO.getJobType() == JobType.SYNC.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_SYNC);
    } else if (jobDTO.getJobType() == JobType.TRANSFER.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_TRANSFER);
    } else if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_SEMISTRUCTURE);
    } else if (jobDTO.getJobType() == JobType.STREAM.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_STREAM);
    } else if (jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_API);
    } else {
      logger.debug("不支持");
    }

    definitionMap.put(ProcessConstants.REMARK, "datax任务");
    definitionMap.put(ProcessConstants.DATASOURCE_ID, 1);
    definitionMap.put(ProcessConstants.CONNECTS, jsonMap.get(ProcessConstants.CONNECTS));
    definitionMap.put(ProcessConstants.LOCATIONS, Constants.STRING_BLANK);
    definitionMap.put(ProcessConstants.NAME, jobDTO.getName());

    // 默认下线状态
    definitionMap.put(ProcessConstants.RELEASE_STATE, 0);
    definitionMap.put(ProcessConstants.PROCESS_DEFINITION_JSON, jsonMap.get(ProcessConstants.PROCESS_DEFINITION_JSON));
    definitionMap.put(ProcessConstants.EXTRA_PARAMS, jsonMap.get(ProcessConstants.EXTRA_PARAMS));

    try {
      processDefinitionApi.createProcessDefinition(definitionMap);
      return HttpStatus.SC_OK;
    } catch (Exception e) {
      logger.error(String.format("调用DS生成流程定义报错,jobId is %s", jobId));
      throw new BusinessException(
          String.format("调用DS生成流程定义报错,jobId is %s,msg:%s", jobId, e.getMessage()));
    }
  }

  /** 修改DS流程定义 */
  public Integer updateProcessDefinition(String jobId) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    // 1、构建Dolphinscheduler流程定义json数据和流程定义节点连接信息json
    Map<String, String> jsonMap = new HashMap<>();
    if (jobDTO.getJobType() == JobType.TRANSFER.getCode()
        || jobDTO.getJobType() == JobType.SYNC.getCode()) {
      // 应用数据采集
      jsonMap = structureTaskJsonService.buildDsJson(jobDTO);
    } else if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      // 文件采集
      jsonMap = fileTaskJsonService.buildDsJson(jobDTO);
    } else if (jobDTO.getJobType() == JobType.STREAM.getCode()) {
      // 流式采集
      jsonMap = streamTaskJsonService.buildDsJson(jobDTO);
    } else if (jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
      // API采集
      jsonMap = apiTaskJsonService.buildDsJson(jobDTO);
    } else {
      logger.warn(DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + jobDTO.getJobType());
    }

    // 2、调用DS服务保存流程定义
    Map<String, Object> definitionMap = new HashMap<>();
    definitionMap.put(ProcessConstants.ID, jobId);
    definitionMap.put(ProcessConstants.PROJECT_ID, jobDTO.getProjectId());
    definitionMap.put(ProcessConstants.TYPE_ID, TaskType.DATAX.getDescription());

    if (jobDTO.getJobType() == JobType.SYNC.getCode()
        || jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()
        || jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_SYNC);
    } else if (jobDTO.getJobType() == JobType.TRANSFER.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_TRANSFER);
    } else if (jobDTO.getJobType() == JobType.STREAM.getCode()) {
      definitionMap.put(TYPE, ProcessConstants.JOB_STREAM);
    } else {
      logger.debug("不支持");
    }

    definitionMap.put(ProcessConstants.REMARK, "datax任务");
    definitionMap.put(ProcessConstants.DATASOURCE_ID, 1);
    definitionMap.put(ProcessConstants.CONNECTS, jsonMap.get(ProcessConstants.CONNECTS));
    definitionMap.put(ProcessConstants.LOCATIONS, Constants.STRING_BLANK);
    definitionMap.put(ProcessConstants.NAME, jobDTO.getName());

    // 默认下线状态
    definitionMap.put(ProcessConstants.RELEASE_STATE, 0);
    definitionMap.put(ProcessConstants.PROCESS_DEFINITION_JSON, jsonMap.get(ProcessConstants.PROCESS_DEFINITION_JSON));
    definitionMap.put(ProcessConstants.EXTRA_PARAMS, jsonMap.get(ProcessConstants.EXTRA_PARAMS));
    boolean success = processDefinitionApi.updateProcessDefinition(definitionMap);

    if (success) {
      return HttpStatus.SC_OK;
    } else {
      logger.error(String.format("调用DS修改流程定义报错,jobId is %s", jobId));
      throw new BusinessException(String.format("调用DS修改流程定义报错,jobId is %s", jobId));
    }
  }

  public String getSourceColumn(TaskDTO task) {
    List<String> sourceColumnList = new ArrayList<>();
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMapping : columnMappingDtoList) {
        sourceColumnList.add(columnMapping.getSrcColumnName());
      }
    }

    String sourceColumn = String.join(Constants.COMMA, sourceColumnList);
    return sourceColumn;
  }

  public String getDstColumn(TaskDTO task) {
    List<String> dstColumnList = new ArrayList<>();
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMapping : columnMappingDtoList) {
        dstColumnList.add(columnMapping.getDstColumnName());
      }
    }

    return String.join(Constants.COMMA, dstColumnList);
  }

  /**
   * splitIntoBlocks函数将列表分成多个子列表
   * @param <T>
   * @return
   */
  public static <T> List<List<T>> splitIntoBlocks(List<T> originalList) {
    double num = originalList.size() / COLLECT_NUM;
    int blockSize = (int) Math.ceil(num);

    List<List<T>> result = new ArrayList<>();

    for (int startIndex = 0; startIndex < originalList.size(); startIndex += blockSize) {
      int endIndex = Math.min(startIndex + blockSize, originalList.size());
      result.add(originalList.subList(startIndex, endIndex));
    }

    return result;
  }

  public static void main(String[] args) {
    List<Integer> numbers = new ArrayList<>();
    for (int i = 1; i <= 3; i++) {
      numbers.add(i);
    }

    List<List<Integer>> blocks = splitIntoBlocks(numbers);

    System.out.println("原始列表：" + numbers);
    System.out.println();

    for (int i = 0; i < blocks.size(); i++) {
      System.out.print("第 " + (i+1) + " 个子列表：");

      if (!blocks.get(i).isEmpty()) {
        System.out.println(blocks.get(i));
      } else {
        System.out.println("<空>");
      }
    }
  }

}
